package com.lagou.rabbitmq.publisherconfirm;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

public class PublisherConfirmProducer3 {

    public static void main(String[] args) throws Exception {

        ConnectionFactory factory = new ConnectionFactory();
        //指定协议 amqp://
        //指定用户名 root
        //指定密码 123456
        //指定host 47.115.74.219
        //指定端口号 5672
        //指定虚拟主机 %2f
        factory.setUri("amqp://root:123456@47.115.74.219:5672/%2f");

        Connection connection = factory.newConnection();

        Channel channel = connection.createChannel();

        //向RabbitMQ服务器发送AMQP命令，将当前通道标记为发送方确认通道
        AMQP.Confirm.SelectOk selectOk = channel.confirmSelect();

        channel.queueDeclare("queue.pc",true,false,false,null);
        channel.exchangeDeclare("ex.pc", BuiltinExchangeType.DIRECT,true,false,null);
        channel.queueBind("queue.pc","ex.pc","key.pc");

        ConcurrentNavigableMap<Long,String> outStandingConfirms = new ConcurrentSkipListMap<>();

        ConfirmCallback clearOutstandingConfirms = new ConfirmCallback() {
            @Override
            public void handle(long l, boolean b) throws IOException {
                if(b){
                    System.out.println("编号小于等于" + l + "的消息已经被确认了");
                    ConcurrentNavigableMap<Long, String> headMap = outStandingConfirms.headMap(l, true);
                    //清空outstandingConfirms中已经被确认的消息信息
                    headMap.clear();
                }else {
                    outStandingConfirms.remove(l);
                    System.out.println("编号为：" + l + "消息被确认");
                }
            }
        };

        channel.addConfirmListener(clearOutstandingConfirms,(delivery,multiple) -> {
            if(multiple){
                //将没有确认的消息记录到一个集合中
                //此处省略
                System.out.println("消息编号小于等于：" + delivery + "的消息不确认");
            }else {
                System.out.println("消息编号为：" + delivery + "的消息不确认");
            }
        });

        String message = "hello-";
        for(int i = 0; i < 100;i++){
            long nextPublishSeqNo = channel.getNextPublishSeqNo();
            channel.basicPublish("ex.pc","key.pc",null,(message + i).getBytes(StandardCharsets.UTF_8));
            System.out.println("消息已经发送还未确认");
            outStandingConfirms.put(nextPublishSeqNo,(message + i));
        }

        //等待消息被确认
        Thread.sleep(10000);

        channel.close();
        connection.close();

    }

}
